package com.syg.kafka02;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author syg
 */
public class Producer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 配置参数
        Properties properties = new Properties();
        // bootstrap.servers：broker的地址清单
        properties.put("bootstrap.servers", "localhost:9092");
        // key.serializer：消息的键的序列化器
        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        // value.serializer：消息的值的序列化器
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        // 2. 根据参数创建KafkaProducer实例（生产者）
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        // 3. 创建ProducerRecord实例（消息）
        ProducerRecord<String, String> record = new ProducerRecord<>("topic-demo", "hello kafka");
        // 4. 发送消息
        Future<RecordMetadata> sendFuture = producer.send(record);
        // send方法的返回值是一个Future对象，当调用其get方法时将阻塞等待Kafka的响应
        RecordMetadata recordMetadata = sendFuture.get();
        // RecordMetadata对象中包含有消息的一些元数据，如消息的主题、分区号、分区中的偏移量、时间戳等。
        // 5. 关闭生产者示例
        producer.close();
    }
}
